#include <string.h>

#include "hj212_client.h"
#include "hj212_msg.h"
#include "hj212_client_ops.h"
#include "hj212_pack.h"

#define HJ_LOG_TAG  "hj212.stream"      // 日志标签
#define HJ_LOG_LVL HJ_DBG_LOG           // 日志等级
#include "hjdebug.h"

static inline int hj212_datasource_is_empty(struct hj212_datasource *source)
{
    if (source->is_empty)
        return source->is_empty(source);
    HJ_LOG_W("There is no (.is_empty) element in the datasource");
    return 1;
}

static inline int hj212_datasource_next(struct hj212_datasource *source, struct hj212_datamsg *msg)
{
    if (source->next)
        return source->next(source, msg);
    HJ_LOG_W("There is no (.next) element in the datasource");
    return -1;
}

static inline struct hj212_datamsg *hj212_datasource_references(struct hj212_datasource *source)
{
    HJ_ASSERT(source->references);
    return source->references(source);
}

static inline void hj212_datasource_dereference(struct hj212_datasource *source, struct hj212_datamsg *msg)
{
    if (source->dereference)
        source->dereference(source, msg);
}

int get_hj212_datastream_state(struct hj212_datastream *stream)
{
    enum hj212_response_state result = HJ212_DATASTREAM_BUSY;

    if (hj212_response_is_idle(&stream->response))  // 没有等待响应的状态
    {
        if (gen_blk_queue_block_num(&stream->msg->queue) == 0 && // 数据块队列和数据源都没数据则数据流空
                hj212_datasource_is_empty(stream->source))
        {
            result = HJ212_DATASTREAM_EMPTY;
        }
        else        // 否则空闲
        {
            result = HJ212_DATASTREAM_IDLE;
        }
    }

    return result;
}

static void hj212_datastream_pack_finish(struct hj212_datastream *stream)
{
    stream->count = 0;
    gen_blk_queue_pop(&stream->msg->queue, NULL, stream->itemnum);
    /* 包号自增 */
    stream->msg->pno++;

    if (gen_blk_queue_block_num(&stream->msg->queue) == 0)
    {
        hj212_datasource_dereference(stream->source, stream->msg);
        stream->msg = NULL;
    }
}

static int hj212_datastream_response_match(struct hj212_response *res, struct hj212_generalmsg *msg, uint32_t package_state, void *param)
{
    if (package_state & ~UNPACK_NO_CP)  // 忽略CP字段，其他错误一概退出
    {
        return -1;
    }
    if (msg->parent.cn != res->cn) // 比较CN
    {
        return -1;
    }
    if (strncmp(msg->parent.qn, res->qn, HJ212_QN_MAX_LEN))    // 比较QN
    {
        return -1;
    }
#if 0
    struct hj212_datastream *stream = param;
    stream->count = 0;
    gen_blk_queue_pop(&stream->msg->queue, NULL, stream->itemnum);
    /* 包号自增 */
    stream->msg->pno++;
#endif
    hj212_datastream_pack_finish(param);
    return 0;
}

static int hj212_datastream_response_timeout(struct hj212_response *res, void *param)
{
    hj212_response_unregister(res);
    struct hj212_datastream *stream = param;

    if (++stream->count >= stream->maxret)
    {
#if 0
        stream->count = 0;
        gen_blk_queue_pop(&stream->msg->queue, NULL, stream->itemnum);
        /* 包号自增 */
        stream->msg->pno++;
#endif
        hj212_datastream_pack_finish(stream);
    }

    stream->event_valid = 0;    // 标记事件无效

    /* 更新时间戳, 然后向系统push一个流传输超时事件 */
    stream->transfer_event.parent.timestamp = hj212_client_current_timestamp(stream->owner);
    stream->transfer_event.parent.type = HJ_EVENT_STREAM_TRANSFER_TIMEOUT;
    hj212_client_event_push(stream->owner, &stream->transfer_event.parent);

    return 0;
}

void datastream_send_callback(struct hj212_client *client, void *p)
{
    struct hj212_datastream *stream = p;
    if (stream->event_valid)
    {
        /* 更新时间戳, 然后向系统push一个流传输事件 */
        stream->transfer_event.parent.timestamp = hj212_client_current_timestamp(client);
        hj212_client_event_push(client, &stream->transfer_event.parent);
        stream->event_valid = 0;
    }
}

int hj212_datastream_push(struct hj212_datastream *stream, hj212_client_t dst, int is_response)
{
    hj212_client_t client = dst;

    if (!stream->msg)
    {
        if (!(stream->msg = hj212_datasource_references(stream->source)))
        {
            return -1;
        }
    }

    if (gen_blk_queue_block_num(&stream->msg->queue) == 0)   // 暂存的数据没了，重新从外部取数据
    {
        if (hj212_datasource_is_empty(stream->source))
        {
            return -1;
        }

        memset(stream->msg->parent.qn, 0, HJ212_QN_MAX_LEN);
        hj212_client_command(client, GET_HJ212_SYS_CODE, stream->msg->parent.st);
        hj212_client_command(client, GET_HJ212_ACCESS_PWD, stream->msg->parent.access_pwd);
        hj212_client_command(client, GET_HJ212_CLIENT_MN, stream->msg->parent.mn);
        stream->msg->parent.cn = -1;
        stream->msg->parent.flag = -1;
        stream->type = stream->msg->type;
        if (hj212_datasource_next(stream->source, stream->msg) < 0)
        {
            HJ_LOG_W("#%d Abnormal datasource (type=%d)", client->id, stream->type);
            return -1;
        }
        if (stream->msg->parent.cn < 0)
        {
            stream->msg->parent.cn = hj212_datamsg_cn(stream->type);
        }
        if (stream->msg->parent.flag < 0)
        {
            int protocol = HJ212_PROTOCOL_2017;
            hj212_client_command(client, GET_HJ212_PROTOCOL, &protocol);
            stream->msg->parent.flag =
                protocol == HJ212_PROTOCOL_2017 ?
                HJ_FLAG_HJ212_2017 : HJ_FLAG_HJ212_2005;
            stream->msg->parent.flag |= HJ_FLAG_PACK_PNUM;
            if (is_response)
            {
                stream->msg->parent.flag |= HJ_FLAG_NEED_RESPONSE;
            }
        }
        stream->msg->pno = 0;
        int pol_maxnum = hj212_datamsg_item_maxnum_per_package(stream->msg->type);
        int pol_num = gen_blk_queue_block_num(&stream->msg->queue);
        stream->msg->pnum = pol_num / pol_maxnum;
        if (stream->msg->pnum * pol_maxnum < pol_num)
        {
            stream->msg->pnum++;
        }
    }
    if (gen_blk_queue_block_num(&stream->msg->queue) > 0)
    {
        /* 拿到缓冲区 */
        struct send_buffer *buffer = hj212_acquire_sendbuf(client);
        if (!buffer)
        {
            // 发送缓冲区满
            return -1;
        }

        /* 拿到参数 */
        int protocol = HJ212_PROTOCOL_2017;
        hj212_client_command(client, GET_HJ212_PROTOCOL, &protocol);
        struct timeval time = hj212_client_current_subsec(client);
        /* 打包 */
        memset(buffer->mempool, 0, HJ212_PACK_MAX_SIZE);
        if (stream->count == 0) // 第一次打包这条数据重新获取最大数量
        {
            stream->itemnum = hj212_datamsg_item_maxnum_per_package(stream->msg->type);
        }
        stream->itemnum = hj212_datamsg_pack(stream->msg, protocol, time,
                                             stream->itemnum,
                                             buffer->mempool, HJ212_PACK_MAX_SIZE);

        /* 设置一个事件, 发送时将这个事件push到系统 */
        stream->transfer_event.parent.type = HJ_EVENT_STREAM_TRANSFER;
        stream->transfer_event.parent.timestamp = 0;    // 设置为0, 将时间戳和事件输出交给send_callback做
        stream->transfer_event.datatime = stream->msg->datatime;
        stream->transfer_event.cn = stream->msg->parent.cn;
        stream->event_valid = 1;    // 标记事件有效

        /* 响应相关处理 */
        if (is_response)
        {
            int maxret = 3;
            hj212_client_command(client, GET_HJ212_MAXRET, &maxret);
            stream->maxret = maxret;

            int timeout = 5;
            hj212_client_command(client, GET_HJ212_STREAM_TIMEOUT, &timeout);
            hj212_response_register_default(client, &stream->response, timeout, hj212_datamsg_response_cn(stream->msg->type), stream->msg->parent.qn);
        }
        else
        {
#if 0
            stream->count = 0;
            gen_blk_queue_pop(&stream->msg->queue, NULL, stream->itemnum);
            /* 包号自增 */
            stream->msg->pno++;
#endif
            hj212_datastream_pack_finish(stream);
        }
        hj212_append_sendbuf(client, buffer, datastream_send_callback, stream);
        return 0;
    }
    return -1;  // 正常情况不应该到这里
}

struct hj212_datastream *hj212_datastream_create(struct hj212_client *client, int type, struct hj212_datasource *source)
{
    struct hj212_datastream *result = NULL;
    struct hj212_datastream *stream = NULL;

    if (!source)
        goto __exit;
    if (type < HJ212_MSG_TYPE_RTD && type > HJ212_MSG_TYPE_NUM)
    {
        goto __exit;
    }
    int typesize = hj212_datamsg_typesize(type);
    if (typesize < 0)
        goto __exit;
    stream = hj212_client_calloc(client, 1, sizeof(*stream));
    if (!stream)
        goto __exit;

    stream->msg = NULL;
    stream->owner = client;
    stream->source = source;
    stream->type = type;
    /* 响应控制块初始化 */
    hj212_response_init(&stream->response,
                        hj212_datastream_response_match, stream,
                        hj212_datastream_response_timeout, stream);

    result = stream;
    HJ_LOG_I("#%d datastream(%d) create succeed", client->id, stream->type);
__exit:
    if (!result)
    {
        if (stream)
            hj212_client_free(client, stream);
        HJ_LOG_W("#d datastream(%d) create failure", client->id, stream->type);
    }
    return result;
}

void hj212_datastream_delete(struct hj212_client *client, struct hj212_datastream *stream)
{
    // hj212_client_free(client, stream->msg.queue.mempool);
    hj212_client_free(client, stream);
}

























